from __future__ import with_statement

import sys
import os
import shutil
import unittest
import datetime
import traceback
import server
from kineta import *


class BasicTests(unittest.TestCase):

    def setUp(self):
        self._dbdir = "db"
        if os.path.isdir(self._dbdir):
            shutil.rmtree(self._dbdir)
        self.open()

    def _open(self):
        db = FileDB(self._dbdir)
        db.set_bg_error_cb(self.bgerror)
        return db

    def open(self):
        self.db = self._open()

    def bgerror(self, exc_info):
        traceback.print_exception(*exc_info)

    def testCRUD(self):
        self.db.insert("people",
            [{"id": {"identity": True}, "name": "will", "city": "roswell"},
             {"id": {"identity": True}, "name": "ben", "city": "atlanta"}])
        with self.db.select(table = ["people", "alias"],
                       fields = ["id", "name", "city"],
                       where = {"name": ("between", "a", "d"),
                                 "city": "atlanta",
                                 "id": ("!=", None)},
                       order = "id") as reader:
            self.assertEqual(reader.field_names, [["alias", "id"],
                                                  ["alias", "name"],
                                                  ["alias", "city"]])
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], [2L, "ben", "atlanta"])
        num = self.db.update("people", {"name": "bob"},
                                       {"name": ("==", "ben")})
        self.assertEqual(num, 1)
        with self.db.select(table = "people",
                       fields = ["id", "name", "city"],
                       where = {"name": ("between", "a", "d"),
                                 "city": ("==", "atlanta"),
                                 "id": ("!=", None)},
                       order = "id") as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], [2L, "bob", "atlanta"])
        num = self.db.delete("people", {"name": ["bob"]})
        self.assertEqual(num, 1)
        with self.db.select(table = "people", fields = ["name"]) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], ["will"])

    def testCloseAndReOpen(self):
        self.db.insert("x", {"a": 1, "b": 2})
        self.db.close()
        with self._open() as db:
            with db.select(table = "x", fields = ["a", "b"]) as reader:
                res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], [1, 2])
        self.db = self._open()

    def testMissingColTable(self):
        self.db.insert("x", {"a": 1, "b": 2})
        plan = self.db.explain(table = "x", fields = ["a", "b", "c"])
        self.assertEqual(plan['strategy'], 'table_scan')
        self.assertEqual(plan['table'], 'x')
        with self.db.select(table = "x", fields = ["a", "b", "c"]) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], [1, 2, None])
        with self.db.select(table = "x", fields = ["c", "d"]) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], [None, None])
        plan = self.db.explain(table = "y", fields = ["c", "d"])
        self.assertEqual(plan['strategy'], 'empty')
        with self.db.select(table = "y", fields = ["c", "d"]) as reader:
            res = list(reader)
        self.assertEqual(len(res), 0)
        with self.db.select(table = "x", fields = [None]) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], [None, ])
        num = self.db.update("y", {"a": 1})
        self.assertEqual(num, 0)
        self.assert_("y" not in self.db.list_tables())
        num = self.db.delete("y")
        self.assertEqual(num, 0)
        self.assert_("y" not in self.db.list_tables())

    def testOrder(self):
        self.db.insert_bulk("x", ("a", "b"),
                            [[1, 'g'], [1, 'f'], [5, 'e'], [100, 'e']])
        with self.db.select(table = "x", fields = ("a", "b"),
                            order = [("a", "desc"), ("b", "asc")]) as reader:
            res = list(reader)
        self.assertEqual(len(res), 4)
        self.assertEqual(res[0], [100, 'e'])
        self.assertEqual(res[1], [5, 'e'])
        self.assertEqual(res[2], [1, 'f'])
        self.assertEqual(res[3], [1, 'g'])

    def testUnknownWhere(self):
        self.db.insert("x", {"a": 1, "b": 2})
        with self.db.select(table = "x", fields = ("a", "b"),
                            where = {"c": ("==", 3)}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 0)
        with self.db.select(table = "x", fields = ("a", "b"),
                            where = {"c": ("==", None)}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)

    def testNonStdTypes(self):
        self.db.insert("x", {"a": datetime.datetime(2000, 1, 1), "b": [1, 2]})
        with self.db.select(table = "x", fields = ("a", "b"),
                            where = {"b": ("==", [1, 2])}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0][0], str(datetime.datetime(2000, 1, 1)))
        self.assertEqual(res[0][1], [1, 2])
        with self.db.select(table = "x", fields = ("a", "b"),
                            where = {"b": (">", (1, ))}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0][0], str(datetime.datetime(2000, 1, 1)))
        self.assertEqual(res[0][1], [1, 2])
        with self.db.select(table = "x", fields = ("a", "b"),
                            where = {"b": ("<", (3, ))}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0][0], str(datetime.datetime(2000, 1, 1)))
        self.assertEqual(res[0][1], [1, 2])
        with self.db.select(table = "x", fields = ("a", "b"),
                            where = {"b": ("in", (1, 2), (2, 3))}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0][0], str(datetime.datetime(2000, 1, 1)))
        self.assertEqual(res[0][1], [1, 2])

    def testListTables(self):
        self.db.insert("x", {"a": 1})
        self.db.insert("y", {"a": 1})
        self.db.insert("z", {"a": 1})
        tbls = self.db.list_tables()
        self.assertEqual(len(tbls), 3)
        self.assert_("x" in tbls)
        self.assert_("y" in tbls)
        self.assert_("z" in tbls)

    def testListFields(self):
        self.db.insert("x", {"a": 1, "b": 2, "c": 3})
        self.db.insert("x", {"a": 4, "b": 5, "c": 6})
        self.db.insert("x", {"d": 7})
        flds = self.db.list_fields("x")
        self.assertEqual(len(flds), 4)
        self.assert_("a" in flds)
        self.assert_("b" in flds)
        self.assert_("c" in flds)
        self.assert_("d" in flds)

    def testMixTypes(self):
        self.db.insert("x", [{"a": 1},
                             {"a": ("a", "b")},
                             {"a": True},
                             {"a": False},
                             {"a": "test"},
                             {"a": u"test2"}])
        with self.db.select(table = "x", fields = "a", order = "a") as reader:
            res = list(reader)
        self.assertEqual(len(res), 6)
        self.assertEqual(res[0][0], ["a", "b"])
        self.assertEqual(res[1][0], False)
        self.assertEqual(res[2][0], True)
        self.assertEqual(res[3][0], 1)
        self.assertEqual(res[4][0], "test")
        self.assertEqual(res[5][0], u"test2")

    def testUpdateNone(self):
        self.db.insert("x", {"a": 1, "b": 2})
        self.db.update("x", {"b": None})
        with self.db.select(table="x", fields="a",
                            where={"b": ('!=', None)}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 0)

    def testDeleteAllDrop(self):
        for i in xrange(10):
            self.db.insert("x", {"a": i})
        num = self.db.delete("x")
        self.assertEqual(num, 10)
        self.db.wait_for_bg_tasks()
        self.assert_("x" in self.db.list_tables())
        self.db.drop_table("x")
        self.assert_("x" not in self.db.list_tables())

        for i in xrange(10):
            self.db.insert("x", {"a": i})
        self.db.create_index("x", "a")
        self.db.wait_for_bg_tasks()
        num = self.db.delete("x", {"a": ('<', 10)})
        self.assertEqual(num, 10)
        self.db.wait_for_bg_tasks()
        self.assert_("x" in self.db.list_tables())
        self.db.drop_table("x")
        self.assert_("x" not in self.db.list_tables())

    def testPaging(self):
        for i in xrange(10):
            self.db.insert("x", {"a": i})
        reader = self.db.select("x", fields=["a"], order=["a"])
        for i in xrange(3):
            row = reader.next()
            self.assertEqual(i, row[0])
        reader.close()

    def testAsDict(self):
        self.db.insert("x", {"a": 1, "b": 2})
        with self.db.select("x", fields=["a", "b"]).as_dict() as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], {"x": {"a": 1, "b": 2}})

    def testStringOps(self):
        self.db.insert("x", {"a": "abcdefg"})
        self.db.insert("x", {"a": "hijklmnop"})
        self.db.insert("x", {"a": "qrstuvwxyz"})

        with self.db.select("x", fields=["a"],
                            where={"a": ('startswith', 'hij')}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], ["hijklmnop"])

        with self.db.select("x", fields=["a"],
                            where={"a": ('contains', 'jklm')}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], ["hijklmnop"])

        with self.db.select("x", fields=["a"],
                            where={"a": ('endswith', 'p')}) as reader:
            res = list(reader)
        self.assertEqual(len(res), 1)
        self.assertEqual(res[0], ["hijklmnop"])

    def tearDown(self):
        self.db.close()


class BasicTestsRemote(BasicTests):

    def _open(self):
        return RemoteDB("http://localhost:20000/db")

    def open(self):
        self.server = server.TestServer()
        self.server.app.databases['db'].set_bg_error_cb(self.bgerror)
        self.server.app.databases['db'].limits.server_row_page_bytes = 2
        self.db = self._open()

    def tearDown(self):
        self.server.quit()
        try:
            self.db.list_tables()
        except:
            pass
        self.db.close()
        self.server.join()
